package com.cloudpivot.project.rabbitmq.service;

import com.cloudpivot.project.rabbitmq.util.ProjectUtil;
import com.rabbitmq.client.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 模拟 每刻4次重试机制 5m 20m 60m error
 */
@Component
public class MayCurContinueTryConsumerService {
    @Value("${rabbitmq-config.host-name}")
    private String HOST_NAME;
    @Value("${rabbitmq-config.host-port}")
    private Integer HOST_PORT;
    @Value("${rabbitmq-config.username}")
    private String USER_NAME;
    @Value("${rabbitmq-config.password}")
    private String PASSWORD;
    //    private static final String QUEUE_NAME = "first_queues";
    private static final String DLX_EXCHANGE = "dlx_exchange";
    private static final String NORMAL_QUEUE = "normal_queue";
    private static final String MAY_CUR_VIRTUAL_HOST = "/maycur_business";


    /**
     * 每刻business对应多次重复 消费者
     */
    @Bean
    public void maycurContinueCunsumer() throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(HOST_NAME);
        factory.setPort(HOST_PORT);
        factory.setUsername(USER_NAME);
        factory.setPassword(PASSWORD);
        factory.setVirtualHost(MAY_CUR_VIRTUAL_HOST);
        Connection rabbitMQConnection = factory.newConnection();
        Channel channel = rabbitMQConnection.createChannel();
        channel.basicQos(1);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received message: " + message);

            try {
                //正常处理业务
                //todo 正常处理方法
                throw new Exception("模拟处理失败");
            } catch (Exception e) {
                System.out.println("处理失败: " + e.getMessage());

                AMQP.BasicProperties properties = delivery.getProperties();
                Map<String, Object> headers = properties.getHeaders() != null ? properties.getHeaders() : new HashMap<>();

                int retryCount = (int) headers.getOrDefault("x-retry-count", 0);

                if (retryCount < 1) {
                    // 第一次失败，重新发布到retry_queue_30s
                    headers.put("x-retry-count", retryCount + 1);
                    headers.put("x-death-reason", "first_retry");

                    AMQP.BasicProperties newProperties = new AMQP.BasicProperties.Builder()
                            .headers(headers)
                            .build();

                    channel.basicPublish(DLX_EXCHANGE, "retry_5m", newProperties, delivery.getBody());
                    System.out.println("消息已重新发布到retry_queue_5m===>"+ProjectUtil.getNowTime());
                } else if (retryCount < 2) {
                    // 第二次失败，重新发布到retry_queue_60s
                    headers.put("x-retry-count", retryCount + 1);
                    headers.put("x-death-reason", "second_retry");

                    AMQP.BasicProperties newProperties = new AMQP.BasicProperties.Builder()
                            .headers(headers)
                            .build();

                    channel.basicPublish(DLX_EXCHANGE, "retry_20m", newProperties, delivery.getBody());
                    System.out.println("消息已重新发布到retry_queue_20m===>>"+ProjectUtil.getNowTime());
                } else if (retryCount < 3){
                    //第三次失败,重新发布到retry_queue_60m
                    headers.put("x-retry-count", retryCount + 1);
                    headers.put("x-death-reason", "three_retry");

                    AMQP.BasicProperties newProperties = new AMQP.BasicProperties.Builder()
                            .headers(headers)
                            .build();

                    channel.basicPublish(DLX_EXCHANGE, "retry_60m", newProperties, delivery.getBody());
                    System.out.println("消息已重新发布到retry_queue_60m===>>"+ProjectUtil.getNowTime());
                }else {
                    // 所有重试失败，发布到error_queue
                    channel.basicPublish(DLX_EXCHANGE, "error", null, delivery.getBody());
                    System.out.println("消息已发送到error_queue===>>"+ProjectUtil.getNowTime());
                }

                // 确认消息已被处理
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        };

            channel.basicConsume(NORMAL_QUEUE, false, deliverCallback, consumerTag -> {
        });

    }

}
